Flink源码|自定义Format消费MaxwellCDC数据

其实并不复杂：先通过 jsonDeserializer 将字节数组根据 [data: ROW, old: ROW, type: String] 的 schema 反序列化成 RowData，然后根据 “type” 列的值来判断数据是什么类型：增、改、删；再根据数据类型取出 “data” 或者 “old” 区的数据，来组装成 Flink 认识的 INSERT/DELETE/UPDATE 数据并下发。

对象 jsonDeserializer 即 JSON 格式的反序列化器，它可以通过指定的 RowType 类型，读取 JSON 的字节数组中指定的字段并反序列化成 RowData。在我们的场景中，我们需要去读取如下 Maxwell 数据的 “data”, “old” 和 “type” 部分的数据。

{"database":"test","table":"product","type":"update","ts":1596684928,"xid":7291,"commit":true,"data":{"id":102,"name":"car battery","description":"12V car battery","weight":5.17},"old":{"weight":8.1}}

因此 MaxwellJsonDeserializationSchema 中定义的 JSON 的 RowType 如下所示。

private RowType createJsonRowType(DataType databaseSchema) {
       // Maxwell JSON contains other information, e.g. "database", "ts"
       // but we don't need them
       return (RowType) DataTypes.ROW(
           DataTypes.FIELD("data", databaseSchema),
           DataTypes.FIELD("old", databaseSchema),
           DataTypes.FIELD("type", DataTypes.STRING())).getLogicalType();
   }

databaseSchema 是用户通过 DDL 定义的 schema 信息，也对应着数据库中表的 schema。结合上面的 JSON 和代码，我们能够得知 jsonDeserializer 只会取走 byte[] 中 data、old、type 这三个字段对应的值，其中 data 和old 还是个嵌套JSON，它们的 schema 信息和 databaseSchema 一致。由于 Maxwell 在同步数据时，“old”区不包含未被更新的字段，所以 jsonDeserializer 返回后，我们会通过 “data” 区的 RowData 将 old 区的缺失字段补齐。

得到 RowData 之后，会取出 type 字段，然后根据对应的值，会有三种分支：

• insert：取出 data 中的值，也就是我们通过DDL定义的字段对应的值，再将其标记为 RowKind.INSERT 类型数据，最后下发。
• update：分别取出 data 和 old 的值，然后循环 old 中每个字段，字段值如果为空说明是未修改的字段，那就用 data 中对应位置字段的值替代；之后将 old 标记为 RowKind.UPDATE_BEFORE 也就意味着 Flink 引擎需要将之前对应的值撤回，data 标记为 RowKind.UPDATE_AFTER 正常下发。
• delete：取出 data 中的值，标记为 RowKind.DELETE，代表需要撤回。

处理的过程中，如果抛出异常，会根据 DDL 中maxwell-json.ignore-parse-errors的值来确定是忽视这条数据继续处理下一条数据，还是让任务报错。

笔者在 maxwell-json 反序列化功能的基础之上，还实现了序列化的功能，即能将 Flink 产生的 changelog 以 Maxwell 的 JSON 格式输出到外部系统中。其实现思路与反序列化器的思路正好相反，更多细节可以参考 Pull Request 中的实现。

PR 实现详情链接： 
https://github.com/apache/flink/pull/13090

功能演示

给大家演示一下从 Kafka 中读取 Maxwell 推送来的 maxwell json 格式数据，并将聚合后的数据再次写入 Kafka 后，重新读出来验证数据是否正确。

Kafka 数据源表

CREATE TABLE topic_products (
 -- schema is totally the same to the MySQL "products" table
 id BIGINT,
 name STRING,
 description STRING,
 weight DECIMAL(10, 2)
) WITH (
'connector' = 'kafka',
'topic' = 'maxwell',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'testGroup',
'format' = 'maxwell-json');

Kafka 数据结果表&数据源表

CREATE TABLE topic_sink (
 name STRING,
 sum_weight DECIMAL(10, 2)
) WITH (
'connector' = 'kafka',
'topic' = 'maxwell-sink',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'testGroup',
'format' = 'maxwell-json'
);

MySQL 表

-- 注意，这部分 SQL 在 MySQL 中执行，不是 Flink 中的表
CREATE TABLE product (
id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255),
description VARCHAR(512),
weight FLOAT
);
truncate product ;
ALTER TABLE product AUTO_INCREMENT = 101;
INSERT INTO product
VALUES (default,"scooter","Small 2-wheel scooter",3.14),
      (default,"car battery","12V car battery",8.1),
      (default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8),
      (default,"hammer","12oz carpenter's hammer",0.75),
      (default,"hammer","14oz carpenter's hammer",0.875),
      (default,"hammer","16oz carpenter's hammer",1.0),
      (default,"rocks","box of assorted rocks",5.3),
      (default,"jacket","water resistent black wind breaker",0.1),
      (default,"spare tire","24 inch spare tire",22.2);
UPDATE product SET description='18oz carpenter hammer' WHERE id=106;
UPDATE product SET weight='5.1' WHERE id=107;
INSERT INTO product VALUES (default,"jacket","water resistent white wind breaker",0.2);
INSERT INTO product VALUES (default,"scooter","Big 2-wheel scooter ",5.18);
UPDATE product SET description='new water resistent white wind breaker', weight='0.5' WHERE id=110;
UPDATE product SET weight='5.17' WHERE id=111;
DELETE FROM product WHERE id=111;
UPDATE product SET weight='5.17' WHERE id=102 or id = 101;
DELETE FROM product WHERE id=102 or id = 103;

先看看能不能正常读取 Kafka 中的 maxwell json 数据。

select * from topic_products;

可以看到，所有字段值都变成了 Update 之后的值，同时，被 Delete 的数据也没有出现。

接着让我们再将聚合数据写入 Kafka。

insert into topic_sink select name,sum(weight) as sum_weight from topic_products group by name;

在 Flink 集群的 Web 页面也能够看到任务正确提交，接下来再让我们把聚合数据查出来。

select * from topic_sink

最后，让我们查询一下 MySQL 中的表，来验证数据是否一致；因为在 Flink 中，我们将 weight 字段定义成 Decimal(10,2)，所以我们在查询 MySQL 的时候，需要将 weight 字段进行类型转换。

没有问题，我们的 maxwell json 解析很成功。

写在最后

根据笔者实现 maxwell-json format 的经验，Flink 对于接口的定义、对于模块职责的划分还是很清晰的，所以实现一个自定义 CDC format 非常简单（核心代码只有200多行）。因此，如果你是用的 OGG，或是自研的同步中间件，可以通过本文的思路快速实现一个 CDC format，一起解放你的 CDC 数据！

原文链接
本文为阿里云原创内容，未经允许不得转载。